package cn.itcast.streaming.task;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.sink.SrcDataToHBaseSink;
import cn.itcast.streaming.sink.SrcDataToHBaseSinkOptimizer;
import cn.itcast.streaming.sink.VehicleDetailSinkOptimizer;
import cn.itcast.streaming.utils.JsonParseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

/** 需求：原始数据实时ETL开发
 *   flink实时消费kafka数据，将数据进行转换为javabean,过滤，加工，计算，流式存储等操作。
 */
public class KafkaSourceDataTask extends BaseTask{
    public static void main(String[] args) {

        /**
         *  1.初始化flink运行环境
         *  2.按照事件时间处理数据
         *  3.开启checkpoint
         *    3.1)每隔30s一个周期开启，时间不能太短，也不能太长（数据丢失）
         *    3.2）设置检查点mode,exactly-noce,保证数据一次性语义
         *    3.2）设置两次checkpoint的时间间隔，避免两次间隔太近，导致频繁checkpoint,出现业务能力下降
         *    3.4）设置超时时间，
         *    3.5）设置最大尝试次数，同一个时间有几个checkpoint在运行
         *    3.6）设置job取消的时候，保留checkpoint计算结果
         *    3.7）作业作业运行过程中，如果checkpoint失败，是否作业也失败
         *    3.8）设置存储后端，使用rocksdb作为状态后端
         *  4.设置重启策略（固定延迟，失败率重启策略，无重启策略）
         *    4.1）默认重启策略：如果开启checkpoint的话，不停的重启，没有开启无重启策略
         *  5.创建flink消费kafka的对象，指定参数信息
         *    5.1）集群地址
         *    5.2）消费者组id
         *    5.3）kafka的分区感知（kafka011版本的粘性机制不需要开启分区感知）
         *    5.4）设置key ,value 反序列
         *    5.5）是否自动提交偏移量
         *    5.6）设置自动递交offset保存到检查点
         *  6.将kafka消费者对象添加到环境中
         *  7.将json字符串转换成javabean
         *  8.过滤正常数据
         *  9.过滤异常数据
         *  10.正常数据的处理
         *    10.1）使用StreamingFileSink将数据实时写入hdfs中
         *    10.2）将正常数据实时写入hbase中
         *  11.异常数据的处理
         *    11.1）使用StreamingFileSink将数据实时写入hdfs中
         *  12.将正常数据处理后常用于分析的字段写入到hbase明细表中
         *  12.启动作业
         */
        StreamExecutionEnvironment env = getEnv(KafkaSourceDataTask.class.getName());

        env.setParallelism(1);//方便测试

        DataStream<String> kafkaStream = createKafkaStream(SimpleStringSchema.class);
        kafkaStream.print("源数据：");
        //todo 7.将json字符串转换成javabean
        SingleOutputStreamOperator<ItcastDataObj> itcastDataObjStream = kafkaStream.filter(
                obj -> StringUtils.isNotEmpty(obj)).map(JsonParseUtil::parseJsonToObject);
        itcastDataObjStream.printToErr("解析后的javabean对象");
        //todo 8.过滤正常数据
        SingleOutputStreamOperator<ItcastDataObj> srcDataStream = itcastDataObjStream.filter(
                itcastDataObj -> StringUtils.isEmpty(itcastDataObj.getErrorData()));
        srcDataStream.print("正常的数据>>>");

        //todo 9.过滤异常数据
        SingleOutputStreamOperator<ItcastDataObj> errorDataStream = itcastDataObjStream.filter(
                itcastDataObj -> StringUtils.isNotEmpty(itcastDataObj.getErrorData()));
        errorDataStream.printToErr("异常的数据>>>");
        //todo 10.正常数据的处理
        //todo 10.1）使用StreamingFileSink将数据实时写入hdfs中
        OutputFileConfig outputFileConfig = OutputFileConfig.builder()
                .withPartPrefix("prefix").withPartSuffix(".ext").build();
        StreamingFileSink srcHdfsSink = StreamingFileSink.forRowFormat(
                        new Path(parameterTool.getRequired("hdfsUri") + "/apps/hive/warehouse/ods.db/itcast_src"),
                        new SimpleStringEncoder<>("utf-8"))
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyyMMdd"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(5))//每5s滚动一次
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(2))//2s内没数据滚动一次
                                .withMaxPartSize(1024 * 1024 * 128)//文件大小到128m滚动
                                .build()
                ).withOutputFileConfig(outputFileConfig).build();
        srcDataStream.map(ItcastDataObj::toHiveString).addSink(srcHdfsSink);
        //todo 10.2）将正常数据实时写入hbase中
        SrcDataToHBaseSink srcDataToHBaseSink = new SrcDataToHBaseSink("itcast_src");
        //srcDataStream.addSink(srcDataToHBaseSink);
        SrcDataToHBaseSinkOptimizer srcDataToHBaseSinkOptimizer = new SrcDataToHBaseSinkOptimizer("itcast_src");
        srcDataStream.addSink(srcDataToHBaseSinkOptimizer);
        //todo 11.异常数据的处理
        //todo 11.1）使用StreamingFileSink将数据实时写入hdfs中
        OutputFileConfig erroroutputFileConfig = OutputFileConfig.builder()
                .withPartPrefix("prefix").withPartSuffix(".ext").build();
        StreamingFileSink errorHdfsSink = StreamingFileSink.forRowFormat(
                        new Path(parameterTool.getRequired("hdfsUri") + "/apps/hive/warehouse/ods.db/itcast_error"),
                        new SimpleStringEncoder<>("utf-8"))
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyyMMdd"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(5))//每5s滚动一次
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(2))//2s内没数据滚动一次
                                .withMaxPartSize(1024 * 1024 * 128)//文件大小到128m滚动
                                .build()
                ).withOutputFileConfig(outputFileConfig).build();
        errorDataStream.map(ItcastDataObj::toHiveString).addSink(errorHdfsSink);

        //todo 12.将正常数据处理后常用于分析的字段写入到hbase明细表中
        VehicleDetailSinkOptimizer vehicleDetailSinkOptimizer = new VehicleDetailSinkOptimizer("itcastsrc_vehicle_detail");
        srcDataStream.addSink(vehicleDetailSinkOptimizer);
        //todo 12.启动作业
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
